/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.fieldmanager;

import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.datanucleus.ClassLoaderResolver;
import org.datanucleus.ObjectManager;
import org.datanucleus.api.ApiAdapter;
import org.datanucleus.exceptions.NucleusDataStoreException;
import org.datanucleus.exceptions.NucleusException;
import org.datanucleus.exceptions.NucleusObjectNotFoundException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.AbstractMemberMetaData;
import org.datanucleus.sco.SCOUtils;
import org.datanucleus.store.ObjectProvider;
import cn.edu.thu.laud.utils.serde.ConverterContext;
import cn.edu.thu.laud.utils.MetaDataUtils;
import org.datanucleus.store.fieldmanager.AbstractFieldManager;

/**
 * 
 * @author zhuoan
 *
 */
public class FetchFieldManager extends AbstractFieldManager
{
	Map<ByteBuffer, ByteBuffer> columns;
	ObjectManager om;
	ConverterContext converterContext;
    ObjectProvider objectProvider;
    Cassandra.Client client;
    AbstractClassMetaData metaData;
    ByteBuffer rowKey;
    String cfName;

    public FetchFieldManager(ObjectManager om, ObjectProvider objcp, ConverterContext converterContext, Cassandra.Client client, AbstractClassMetaData cmd, List<ColumnOrSuperColumn> result, ByteBuffer rowKey, String cfName)
    {
        this.om = om;
        this.objectProvider = objcp;
        this.converterContext = converterContext;
        this.client = client;
        this.metaData = cmd;
        this.rowKey = rowKey;
        this.cfName = cfName;        
        columns = new TreeMap<ByteBuffer, ByteBuffer>();

        for (int index = 0; index < result.size(); index++) {
            ColumnOrSuperColumn columnOrSuperColumn = result.get(index);
            columns.put(columnOrSuperColumn.getColumn().name, columnOrSuperColumn.getColumn().value);
        }
    }
    
    public FetchFieldManager(ObjectManager om, ObjectProvider objcp, ConverterContext converterContext, Cassandra.Client client, Map<ByteBuffer, ByteBuffer> result_map, ByteBuffer rowKey, String cfName)
    {
        this.om = om;
        this.objectProvider = objcp;
        this.converterContext = converterContext;
        this.client = client;
        this.metaData = objcp.getClassMetaData();
        this.rowKey = rowKey;
        this.cfName = cfName;
        this.columns = result_map;
    }

    public FetchFieldManager(ObjectManager om, ObjectProvider objcp, ConverterContext converterContext, Cassandra.Client client, List<ColumnOrSuperColumn> result, ByteBuffer rowKey, String cfName)
    {
        this(om, objcp, converterContext, client, objcp.getClassMetaData(), result,  rowKey, cfName);
    }

    public boolean fetchBooleanField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Boolean) converterContext.getBoolean(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public byte fetchByteField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);
            //value.reset();
			return ByteBufferUtil.getArray(value)[0];

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public char fetchCharField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Character) converterContext.getCharacter(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public double fetchDoubleField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Double) converterContext.getDouble(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public float fetchFloatField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Float) converterContext.getFloat(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public int fetchIntField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Integer) converterContext.getInteger(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public long fetchLongField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Long) converterContext.getLong(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public short fetchShortField(int fieldNumber) {
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (Short) converterContext.getShort(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    public String fetchStringField(int fieldNumber) {
    	
    	try {

			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			ByteBuffer value = this.columns.get(columnName);

			return (String) converterContext.getString(value);

		} catch (Exception e) {
			throw new NucleusException(e.getMessage(), e);
		}
    }

    @SuppressWarnings("unchecked")
	public Object fetchObjectField(int fieldNumber) {
    	       
    	try {
	       // ExecutionContext context = objectProvider.getExecutionContext();
	        ClassLoaderResolver clr = om.getClassLoaderResolver();
	        AbstractMemberMetaData fieldMetaData = metaData
	                .getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
	
	        // handle relations
	        int relationType = fieldMetaData.getRelationType(clr);
	        if(fieldMetaData.isEmbedded() && MetaDataUtils.isRelationSingleValued(relationType)){
	       	 // Persistable object embedded into table of this object
	//           Class<?> embcls = fieldMetaData.getType();
	//           AbstractClassMetaData embcmd = om.getMetaDataManager().getMetaDataForClass(embcls, clr);
	//           if (embcmd != null)
	//           {
	//               // Check for null value (currently need all columns to return null)
	//               // TODO Cater for null (use embmd.getNullIndicatorColumn/Value)                           
	//               ObjectProvider embSM = om.getExecutionContext().newObjectProviderForMember(fieldMetaData, embcmd);
	//               embSM.addEmbeddedOwner(objectProvider, fieldNumber);
	//               FieldManager ffm = new FetchEmbeddedFieldManager(om, embSM, converterContext, fieldMetaData, columns, cfName);
	//               embSM.replaceFields(embcmd.getAllMemberPositions(), ffm);
	//               return embSM.getObject();
	//           }
	           
	           //throw new NucleusUserException("Field " + fieldMetaData.getFullFieldName() + " marked as embedded but no such metadata");
	        	throw new NucleusDataStoreException("Embedded objects are currently unimplemented.");
	   	    }
	        
	        ByteBuffer columnName = MetaDataUtils.getColumnName(metaData, fieldNumber);
	        ByteBuffer value = columns.get(columnName);        
	      
	        if (MetaDataUtils.isRelationSingleValued(relationType)) {   
	
	            if (value == null) {
	                return null;
	            }
	            Object identity = converterContext.getObjectIdentity(om, fieldMetaData.getType(), value);
	            try {
	
					Object object = om.findObject(identity, false, true,
	
					fieldMetaData.getTypeName());
	
					return objectProvider.wrapSCOField(fieldNumber, object,
							false, false, true);
				} catch (NucleusObjectNotFoundException nonfe) {
					// swallow. TODO remove the lazy reference if record is over
					// tombstone time
				}
	
	        } else if (MetaDataUtils.isRelationMultiValued(relationType)) {
	           
	            if (Collection.class.isAssignableFrom(fieldMetaData.getType())) {
	            	
	            	Collection<Object> coll;
					Class<?> elementClass = clr.classForName(fieldMetaData
							.getCollection().getElementType());
					try {
						Class<?> instanceType = SCOUtils
								.getContainerInstanceType(
										fieldMetaData.getType(),
										fieldMetaData.getOrderMetaData() != null);
						coll = (Collection<Object>) instanceType.newInstance();
						
					} catch (Exception e) {
						throw new NucleusDataStoreException(e.getMessage(), e);
					}
	
					// loop through the super columns
	
					// get our list of Strings
	
					ReadCollection columnFetcher = new ReadCollection(client, 
							converterContext, cfName, rowKey, columnName,
							om, elementClass);
	
					// TODO use context.getFetchPlan().getFetchSize()
					columnFetcher.fetchColumns(100, null);
	
					for (Object key : columnFetcher) {
	
						try {
							Object element = om.findObject(key, false,
									true, fieldMetaData.getTypeName());
	
							coll.add(element);
						} catch (NucleusObjectNotFoundException nonfe) {
							// swallow. TODO remove the lazy reference if record
							// is over tombstone time
						}
					}
	
					if (coll.size() == 0) {
						return null;
					}
	
					return objectProvider.wrapSCOField(fieldNumber, coll,
							false, false, true);
	                
	            } else if (Map.class.isAssignableFrom(fieldMetaData.getType())) {
	                // Process all keys, values of the Map that are PC
	            	Map<Object, Object> map;
	
					try {
						Class<?> instanceType = SCOUtils
								.getContainerInstanceType(
										fieldMetaData.getType(),
										fieldMetaData.getOrderMetaData() != null);
						map = (Map<Object, Object>) instanceType.newInstance();
					} catch (Exception e) {
						throw new NucleusDataStoreException(e.getMessage(), e);
					}
	
					ApiAdapter adapter = objectProvider.getExecutionContext()
							.getApiAdapter();
	
					Class<?> keyClass = clr.classForName(fieldMetaData.getMap()
							.getKeyType());
					Class<?> valueClass = clr.classForName(fieldMetaData
							.getMap().getValueType());
	
					Class<?> storedKeyClass = keyClass;
					Class<?> storedValueClass = valueClass;
	
					boolean pcKey = adapter.isPersistable(keyClass);
					boolean pcValue = adapter.isPersistable(valueClass);
	
					if (pcKey) {
						storedKeyClass = converterContext.getKeyClass(
								om,
								fieldMetaData.getMap().getKeyClassMetaData(clr,
										om.getMetaDataManager()));
					}
	
					if (pcValue) {
						storedValueClass = converterContext.getKeyClass(
								om,
								fieldMetaData.getMap().getValueClassMetaData(
										clr, om.getMetaDataManager()));
					}
	
					// TODO use context.getFetchPlan().getFetchSize()
					ReadMap mapReader = new ReadMap(client, converterContext, cfName,
							rowKey, columnName, storedKeyClass,
							storedValueClass);
					mapReader.fetchColumns(100, null);
	
					for (KVEntry entry : mapReader) {
	
						try {
							Object key = null;
	
							if (pcKey) {
	
								key = om.findObject(
										om.newObjectId(keyClass,
												entry.getKey()), false, true,
										fieldMetaData.getTypeName());
							} else {
								key = entry.getKey();
							}
	
							Object val = null;
	
							if (pcValue) {
								val = om.findObject(
										om.newObjectId(valueClass,
												entry.getValue()), false, true,
										fieldMetaData.getTypeName());
							} else {
								val= entry.getValue();
							}
	
							map.put(key, val);
						} catch (NucleusObjectNotFoundException nonfe) {
							// swallow. TODO remove the lazy reference if record
							// is over
							// tombstone time
						}
	
					}
	
					if (map.size() == 0) {
						return null;
					}
	
					return objectProvider.wrapSCOField(fieldNumber, map, false,
							false, true);
	            }else if (fieldMetaData.getType().isArray()) {
	
					ReadMap mapReader = new ReadMap(client, converterContext, cfName,
							rowKey, columnName, Integer.class,
							converterContext.getKeyClass(om, metaData));
					mapReader.fetchColumns(100, null);
	
					int columns = mapReader.getColumnCount();
	
					// TODO make this size more
					Object array = Array.newInstance(fieldMetaData.getType()
							.getComponentType(), columns);
	
					Class<?> elementClass = clr.classForName(fieldMetaData
							.getArray().getElementType());
	
					for (KVEntry entry : mapReader) {
	
						Object id = om.newObjectId(elementClass,
								entry.getValue());
	
						Object element = om.findObject(id, false, true, fieldMetaData.getTypeName());
	
						Array.set(array, (Integer) entry.getKey(), element);
					}
	
					return objectProvider.wrapSCOField(fieldNumber, array,
							false, false, true);
				}
	        }
	
	        // No object defined
			if (value == null) {
				return null;
			}
			
	        return converterContext.getObject(value, this.metaData, fieldNumber);
    	}catch (Exception e){
    		throw new NucleusException(e.getMessage(), e);
    	}
    }

     
    
}